package com.cheng.im.service.massage.service.impl;

import com.cheng.im.common.bean.msg.ProtoMsgV1;
import com.cheng.im.common.cocurrent.FutureTaskScheduler;
import com.cheng.im.common.constants.SeqConstants;
import com.cheng.im.common.entity.SessionCache;
import com.cheng.im.common.protobuiler.MessageAckBuild;
import com.cheng.im.common.protobuiler.MessageReceiveAckBuilder;
import com.cheng.im.service.common.eunms.ConversationTypeEnum;
import com.cheng.im.service.common.resp.CommonResult;
import com.cheng.im.service.conversation.model.req.AddConversationSetReq;
import com.cheng.im.service.conversation.service.ImConversationSetService;
import com.cheng.im.service.message.service.CheckSendMessageService;
import com.cheng.im.service.mq.RocketMQProduce;
import com.cheng.im.service.protobuilder.P2PMessageBuilder;
import com.cheng.im.service.redis.seq.RedisSeq;
import com.cheng.im.service.util.ConversationIdGenerate;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import java.util.Map;

/**
 * @description:
 * @author: lld
 * @version: 1.0
 */
@Service
@Slf4j
public class P2PMessageService {

    @Resource
    CheckSendMessageService checkSendMessageService;

    @Resource
    MessageStoreService messageStoreService;

    @Resource
    ImConversationSetService imConversationSetService;

    @Resource
    RocketMQProduce rocketMQProduce;
    @Resource
    P2PMessageBuilder p2PMessageBuilder;

    @Resource
    RedisSeq redisSeq;

    public void process(ProtoMsgV1.Message pkg){

        log.info("p2p消息开始处理，原始消息：{}", pkg);

        ProtoMsgV1.Message messageFromMessageIdCache = messageStoreService.getMessageFromMessageIdCache
                (pkg.getAppId(), pkg.getMessageRequest().getMsgSdkUuid(), ProtoMsgV1.Message.class);
        if (messageFromMessageIdCache != null){
            FutureTaskScheduler.add(() ->{
                ack(messageFromMessageIdCache);
                //2.发消息给同步在线端
                syncToSender(messageFromMessageIdCache);
                //3.发消息给对方在线端
                Map<String, SessionCache> sessionCacheMap = dispatchMessage(messageFromMessageIdCache);
                if(sessionCacheMap.isEmpty()){
                    //发送接收确认给发送方，要带上是服务端发送的标识
                    receiveAck(messageFromMessageIdCache);
                }
            });
            return;
        }

        if (pkg.getMessageRequest().getFrom().equals(pkg.getMessageRequest().getTo())) {
//            ack(message);
            return;
        }

        //回调
//        CommonResult responseVO = CommonResult.success();
//        if(appConfig.isSendMessageAfterCallback()){
//            responseVO = callbackService.beforeCallback(messageContent.getAppId(), Constants.CallbackCommand.SendMessageBefore
//                    , JSONObject.toJSONString(messageContent));
//        }
//
//        if(!responseVO.isOk()){
//            ack(messageContent,responseVO);
//            return;
//        }

        // appid:messageSeq:2|1
        long seq = redisSeq.doGetSeq(pkg.getAppId() + ":"
                + SeqConstants.P2P_MESSAGE+ ":" + ConversationIdGenerate
                    .generateP2PId(pkg.getMessageRequest().getFrom(), pkg.getMessageRequest().getTo()));

        ProtoMsgV1.Message message = p2PMessageBuilder.updateSeqAndMsgId(pkg, seq);

        FutureTaskScheduler.add(() ->{
            // appId + Seq + (from + to) groupId
            messageStoreService.storeP2PMessage(message);

            // todo: 离线消息
//            OfflineMessageContent offlineMessageContent = new OfflineMessageContent();
//            BeanUtils.copyProperties(messageContent,offlineMessageContent);
//            offlineMessageContent.setConversationType(ConversationTypeEnum.P2P.getCode());
//            messageStoreService.storeOfflineMessage(offlineMessageContent);


            // 更新会话
            AddConversationSetReq addConversationSetReq = new AddConversationSetReq();
            addConversationSetReq.setConversationType(ConversationTypeEnum.P2P.getCode());
            addConversationSetReq.setToId(message.getMessageRequest().getTo());
            addConversationSetReq.setFromId(message.getMessageRequest().getFrom());
            addConversationSetReq.setAppId(message.getAppId());

            imConversationSetService.addConversationSet(addConversationSetReq);

            // 插入数据
            // 1.将消息发送会给自己，因为现在才有消息id那些
            ack(message);
            // 2.发消息给同步在线端
            syncToSender(message);

            // 更新会话
            addConversationSetReq.setFromId(message.getMessageRequest().getTo());
            addConversationSetReq.setToId(message.getMessageRequest().getFrom());

            imConversationSetService.addConversationSet(addConversationSetReq);
            // 3.发消息给对方在线端
            Map<String, SessionCache> stringSessionCacheMap = dispatchMessage(message);

            messageStoreService.setMessageFromMessageIdCache(message.getAppId(),
                    message.getMessageRequest().getMsgSdkUuid(), message);
            if(CollectionUtils.isEmpty(stringSessionCacheMap)){
                //发送接收确认给发送方，要带上是服务端发送的标识
                receiveAck(message);
            }

//            if(appConfig.isSendMessageAfterCallback()){
//                callbackService.callback(messageContent.getAppId(),Constants.CallbackCommand.SendMessageAfter,
//                        JSONObject.toJSONString(messageContent));
//            }
        });
    }

    private Map<String, SessionCache> dispatchMessage(ProtoMsgV1.Message pkg){

        return rocketMQProduce.dispatchMessage(pkg.getMessageRequest().getTo(), pkg);
    }

    private void ack(ProtoMsgV1.Message pkg){

        ProtoMsgV1.Message message = MessageAckBuild.messageAckBuild(pkg, pkg.getMessageRequest().getFrom(),
                pkg.getMessageRequest().getFrom(), pkg.getMessageRequest().getMsgSdkUuid(),
                pkg.getMessageRequest().getMsgId());
        log.info("单聊ack消息为：{}", message);
        rocketMQProduce.sendToUser(message.getMessageAck().getFrom(), message);
    }

    public void receiveAck(ProtoMsgV1.Message pkg){

        ProtoMsgV1.Message message = MessageReceiveAckBuilder.build(pkg, pkg.getMessageRequest().getFrom(),
                pkg.getMessageRequest().getFrom(), pkg.getMessageRequest().getMsgSdkUuid(),
                pkg.getMessageRequest().getMsgId());
        log.info("接收ack: {}", message);
        rocketMQProduce.sendToUser(message.getMessageReceiveAck().getFrom(), message);
    }

    private void syncToSender(ProtoMsgV1.Message pkg){
        rocketMQProduce.sendToUserExceptClient(pkg.getMessageRequest().getFrom(), pkg);
    }

    /**
     * 检查是否允许发送
     * @param fromId
     * @param toId
     * @param appId
     * @return
     */
    public CommonResult imServerPermissionCheck(String fromId,String toId, String appId){
        CommonResult responseVO = checkSendMessageService.checkSenderForvidAndMute(fromId, appId);
        if(!responseVO.isOk()){
            return responseVO;
        }
        responseVO = checkSendMessageService.checkFriendShip(fromId, toId, appId);
        return responseVO;
    }

//    public SendMessageResp send(SendMessageReq req) {
//
//        SendMessageResp sendMessageResp = new SendMessageResp();
//        MessageContent message = new MessageContent();
//        BeanUtils.copyProperties(req,message);
//        //插入数据
//        messageStoreService.storeP2PMessage(message);
//        sendMessageResp.setMessageKey(message.getMessageKey());
//        sendMessageResp.setMessageTime(System.currentTimeMillis());
//
//        //2.发消息给同步在线端
//        syncToSender(message,message);
//        //3.发消息给对方在线端
//        dispatchMessage(message);
//        return sendMessageResp;
//    }
}
